package cn.flink.exercise;

import org.apache.flink.table.api.*;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

/**
 * kafka flink统计
 * 
 * @author mayong
 * @date 2023/04/19
 **/
public class hotGoodsPVFlinkStat {

    public static void main(String[] args) {

        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().build();

        TableEnvironment tableEnvironment = TableEnvironment.create(environmentSettings);

        Schema schema = Schema.newBuilder()
                .column("userId", DataTypes.STRING())
                .column("itemId",DataTypes.STRING())
                .column("categoryId",DataTypes.STRING())
                .column("behavior",DataTypes.STRING())
                .column("timestamp",DataTypes.STRING())
                .build();

        tableEnvironment.createTemporaryTable("sourceTable", TableDescriptor.forConnector("kafka")
                        .schema(schema)
                        .option("topic", "user_behavior")
                        .option("properties.bootstrap.servers","bigdata01:9092,bigdata02:9092,bigdata03:9092")
                        .option("properties.group.id", "behavior")
                        .option("scan.startup.mode", "earliest-offset")
                        .option("format", "json")
                        .build()
        );

        Table table = tableEnvironment.from("sourceTable").select($("itemId"), $("timestamp"));

        Table tableResult = table.window(Slide.over(lit(1).hour()).every(lit(5).minute()).on($("timestamp")).as("w"))
                .groupBy($("itemId"), $("w"))
                .select($("itemId"), $("itemId").count().as("pv"));

        tableResult.execute().print();

    }

}
